package com.ibm.spatiotemp.job;

import java.io.IOException;
import java.util.Collection;
import java.util.HashSet;

import org.apache.log4j.Logger;

import com.ibm.spatiotemp.datamodel.datatype.TemporalLineString;
import com.ibm.spatiotemp.datamodel.datatype.request.TSBuildIdxRequest;
import com.ibm.spatiotemp.job.processor.TSIndexBuildingProcessor;
import com.ibm.spatiotemp.server.config.LocalConfig;
import com.ibm.spatiotemp.source.GPSDataSource;

public class IndexJobRunner {

	public static final int JOBMODE_LOCAL = 0;
	public static final int JOBMODE_SPARK = 1;
	public static final int JOBMODE_MAPREDUCE = 2;

	public static Logger logger = Logger.getLogger(IndexJobRunner.class);

	/**
	 * 
	 * @param id: job id
	 * @param dsrc: datasource(contains multiple time serieses)
	 * @param jobset()
	 * @param jobMode(Job Running mode, @see {@link LocalConfig.IDX_JOBMODE})
	 * @return Collection<TemporalLineString>: a set of linestring splitted, will be used to build up 1stlayer index
	 * @throws Exception
	 */
	public Collection<TemporalLineString> runJobs(JobID id, GPSDataSource dsrc,
			Collection<TSBuildIdxRequest> jobset, int jobMode) throws Exception {
		switch (jobMode) {
		case JOBMODE_LOCAL: {
			logger.info("run index building job for " + id.getJobid()
					+ " locally");
			return runJob_Locally(id, dsrc, jobset);
		}
		case JOBMODE_SPARK: {
			return runJob_Spark(id, dsrc, jobset);
		}
		case JOBMODE_MAPREDUCE: {
			return runJob_MapReduce(id, dsrc, jobset);
		}
		default:
			return runJob_Locally(id, dsrc, jobset);
		}
	}

	protected Collection<TemporalLineString> runJob_Locally(JobID id,
			GPSDataSource dsrc, Collection<TSBuildIdxRequest> jobset)
			throws IOException, InterruptedException {
		Collection<TemporalLineString> linestringset = new HashSet<TemporalLineString>();
		for (TSBuildIdxRequest tsBuildIdxRequest : jobset) {
			TSIndexBuildingProcessor processor = new TSIndexBuildingProcessor(
					dsrc.getTimeseriesByTSIDAndTimeRange(tsBuildIdxRequest));
			linestringset.addAll(processor.buildRtreeIndex(id,
					tsBuildIdxRequest));
		}
		return linestringset;
	}

	protected Collection<TemporalLineString> runJob_Spark(JobID id,
			GPSDataSource dsrc, Collection<TSBuildIdxRequest> jobset) {
		throw new RuntimeException("not implemented");
	}

	protected Collection<TemporalLineString> runJob_MapReduce(JobID id,
			GPSDataSource dsrc, Collection<TSBuildIdxRequest> jobset) {
		
		/*step1: create a new temp folder for Current JOB*/
		
		
		/*step2: write each TSBuildIdxRequest as a line in tempfolder/inputfile*/
		
		
		/*step3: map reduce running job and output reduce result to tempfolder/reduce/ */
		
		
		/*step 4: read from reduce folder and return as Collection<TemporalLineString>*/
		
		
		throw new RuntimeException("not implemented");
	}

}
